/**
 * Copyright (c) 2022 Presto Labs Pte. Ltd.
 * Author: Ju
 */

#pragma once

#include <boost/asio/io_service.hpp>
#include <boost/asio/strand.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/version.hpp>
#include <queue>
#include <set>
#include <string>
#include <thread>

#include "coin2/base/influxdb/influxdb_builder.h"
#include "coin2/base/request/http.h"
#include "coin2/base/scheduler.h"
#include "coin2/base/time.h"

namespace coin2::base::influxdb {

namespace {
static const uint32_t kMaxQueueSize = 4 * 1024;
static const int kMaxFreeHttpClient = 15;
static const int kVersion = 11;
static const uint64_t kOneSec = 1'000'000'000LL;
}  // namespace
namespace http = boost::beast::http;
using coin2::base::request::HttpsConnection;

class InfluxdbClient : public HttpsConnection::Handler {
 public:
  explicit InfluxdbClient(boost::asio::io_context* ioc,
                          std::shared_ptr<Scheduler> timer, std::string host,
                          std::string org, uint16_t port, std::string db,
                          std::string token, uint64_t flush_interval = 1000,
                          uint32_t max_queue_size = kMaxQueueSize);

  ~InfluxdbClient();
  void SendData(std::string buffer);
  void SendData(LineProtocol buffer);
  void Run();

 private:
  void Flush();
  void OnEvent(std::string data);
  int FlushNum(int num);

  // inherited from |HttpsConnection::Handler|
  void OnConnected(std::weak_ptr<HttpsConnection> conn) override;
  void OnRequestSent(std::weak_ptr<HttpsConnection> conn,
                     const RequestType* req) override;
  void OnResponseReceived(std::weak_ptr<HttpsConnection> conn,
                          const ResponseType* rep) override;
  void OnError(std::weak_ptr<HttpsConnection> conn,
               boost::system::error_code ec) override;

  std::shared_ptr<HttpsConnection> TakeClient();
  void RevertClient(std::weak_ptr<HttpsConnection> client);
  void DeferDeleteClient(std::weak_ptr<HttpsConnection> conclientn);

  void DeleteClients();
  void OnTimer();
  void DeleteReq(std::shared_ptr<HttpsConnection> conn);

 private:
  boost::asio::io_context* io_ctx_;
  std::shared_ptr<Scheduler> timer_;
  std::string host_;
  std::string org_;
  int port_;
  std::string target_;
  std::string token_;
  std::queue<std::string> items_;
  std::set<std::shared_ptr<HttpsConnection>> http_clients_;
  std::set<std::shared_ptr<HttpsConnection>> defer_deleted_clients_;

  std::unordered_map<std::shared_ptr<HttpsConnection>, RequestType*> req_map_;

  int64_t flush_ms_;
  int64_t last_flush_time_;
  uint32_t max_queue_size_;

  std::atomic<bool> is_running_;
};
}  // namespace coin2::base::influxdb